package org.apache.spark.examples.sql.streaming;

import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Encoders;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.streaming.StreamingQuery;
import org.apache.spark.sql.streaming.StreamingQueryException;

import java.util.Arrays;

/**
 * Created by admin on 2019/3/23.
 *
 * spark 启动 命令：
 * $ ./bin/run-example org.apache.spark.examples.sql.streaming.JavaStructuredNetworkWordCount localhost 9999
 */
public class JavaStructuredNetworkWordCount {


    public static void main(String[] args){
        System.setProperty("hadoop.home.dir", "C:/hadoop-2.6.0");
        //1.，我们必须导入必要的类并创建一个本地SparkSession，这是与Spark相关的所有功能的起点。
        SparkSession  spark = SparkSession
                .builder()
                .appName("JavaStructuredNetworkWordCount")
                .getOrCreate();
        //2.接下来，让我们创建一个流式DataFrame，它表示从侦听localhost：9999的服务器接收的文本数据，并转换DataFrame以计算字数。
        //socket
        Dataset<Row>  lines = spark
                .readStream()
                .format("socket")
                .option("host","localhost")
                .option("port",9999)
                .load();
        //DataFrame / Dataset是否具有流数据df.isStreaming
        boolean streaming = lines.isStreaming();
        Dataset<String>  words = lines
                .as(Encoders.STRING())  //DataFrame转换为String的数据集.as(Encoders.STRING())
                //我们可以应用flatMap操作将每行拆分为多个单词。结果words数据集包含所有单词
                .flatMap((FlatMapFunction<String,String>)
                        x -> Arrays.asList(x.split(" "))
                                .iterator(), Encoders.STRING());

        Dataset<Row>  wordcounts =  words.groupBy("value").count();

        wordcounts.show();
        //我们将其设置为outputMode("complete")每次更新时将完整的计数集（指定者）打印到控制台。
        // 然后使用启动流式计算start()。
        StreamingQuery query = wordcounts.writeStream()
                .outputMode("complete")
                .format("console")
                .start();
        try {
            //awaitTermination()以防止在查询处于活动状态时退出该进程。
            query.awaitTermination();
        } catch (StreamingQueryException e) {
            e.printStackTrace();
        }

    }
}
